package com.zjl.Kafka.第04章_分布式扩展;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;

import java.util.*;

/**
 * Spark是分布式计算引擎，是一款非常强大的离线分布式计算框架，其中的
 * SparkStreaming模块用于准实时数据处理，其中就可以将Kafka作为数据源进行处理。
 *
 * 需要加依赖
 *
 */
public class F_SparkStreaming分布式计算引擎 {
    public static void main(String[] args) throws InterruptedException {
        SparkConf conf = new SparkConf();//创建配置对象
        conf.setMaster("local[*]");
        conf.setAppName("SparkStreaming");

        //创建环境对象
        JavaStreamingContext ssc = new JavaStreamingContext(conf,new Duration(3*1000));//每3秒去取一次数据

        //使用kafka 最为数据源
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "82.157.71.243:9092");//连接地址
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());//key 的 反序列化
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());//value 的 反序列化
        //消费者组的名称,  kafka 会保存消费者组的偏移量，他优先级最高比其他偏移量高
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "x01");//必选项

        // 默认偏移量为 LEO (每个节点在指定的主题下的最大偏移量)，这样的话，消费者启动晚了，就永远消费不到之前的了
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");//latest ：默认
        //创建消费者对象
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(config);

        //订阅主题
        List<String> topics = new ArrayList<>();
        topics.add("test");
        kafkaConsumer.subscribe(topics);

        JavaInputDStream<ConsumerRecord<String, String>> directStream = KafkaUtils.createDirectStream(
                ssc,
                LocationStrategies.PreferBrokers(),
                ConsumerStrategies.<String, String>Subscribe(topics, config));

        directStream.map((v1) ->{return v1.value();}).print(100);//将数据打印

        ssc.start();//启动

        ssc.awaitTermination();

    }
}
